#include "xlcomm/net/detail/timer_queue.h"

#include <sys/timerfd.h>
#include <unistd.h>

#include "xlcomm/base/logger.h"
#include "xlcomm/net/detail/sockets_ops.h"

namespace xlcomm {
namespace net {

// 创建timerfd文件描述符
int CreateTimerfd() {
  int fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
  if (fd < 0) {
    LOG_SYSFATAL(timerfd_create);
  }
  return fd;
}

// 从现在再过多少timespec才到给定时间
struct timespec HowMuchTimeFromNow(Timestamp when) {
  int64_t microseconds = when - DateTime::NowTimestamp();
  if (microseconds < 100) {
    microseconds = 100;
  }
  struct timespec ts;
  ts.tv_sec =
      static_cast<time_t>(microseconds / DateTime::kMicroSecondsPerSecond);
  ts.tv_nsec = static_cast<long>(
      (microseconds % DateTime::kMicroSecondsPerSecond) * 1000);
  return ts;
}

// 重新设置timerfd到期时间
void ResetTimerfd(int fd, Timestamp when_expire) {
  struct itimerspec new_value;
  MemZero(&new_value, sizeof new_value);
  new_value.it_value = HowMuchTimeFromNow(when_expire);
  if (::timerfd_settime(fd, 0, &new_value, nullptr) < 0) {
    LOG_SYSERROR(timerfd_settime) << "fd:" << fd;
  }
}

// 读timerfd
void ReadTimerfd(int fd) {
  uint64_t data;
  ssize_t n = sockets::read(fd, &data, sizeof(data));
  if (n != sizeof(data)) {
    LOG_SYSERROR(read) << "TimerQueue reads " << n << " bytes instead of 8!";
  }
}

TimerQueue::TimerQueue(EventLoop* loop)
    : loop_(loop), timer_fd_(CreateTimerfd()), timer_channel_(loop, timer_fd_) {
  XLCOMM_ASSERT(loop_ != nullptr);
  LOG_INFO << "TimerQueue created, timer_fd=" << timer_fd_;
  timer_channel_.EnableReading();
  timer_channel_.set_read_callback(std::bind(&TimerQueue::HandleRead, this));
}

TimerQueue::~TimerQueue() {
  timer_channel_.DisableAll();
  timer_channel_.Remove();
  sockets::close(timer_fd_);
  for (const Entry& entry : timers_) {
    delete entry.second;
  }
}

TimerId TimerQueue::AddTimer(TimerCallback cb, Timestamp when_expire,
                             double interval_second) {
  Timer* timer = new Timer(std::move(cb), when_expire, interval_second);
  // 转移到IO线程执行，所以处理过程不必加锁
  loop_->RunInLoop(std::bind(&TimerQueue::AddTimerInLoop, this, timer));
  return TimerId(timer, timer->sequence());
}

void TimerQueue::AddTimerInLoop(Timer* timer) {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  bool earliest_changed = Insert(timer);
  if (earliest_changed) {
    ResetTimerfd(timer_fd_, timer->when_expire());
  }
}

void TimerQueue::HandleRead() {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  ReadTimerfd(timer_fd_);
  LOG_DEBUG << "TimerQueue(timer_fd:" << timer_fd_ << ") read some";
  Timestamp now = DateTime::NowTimestamp();
  std::vector<Entry> expired = GetExpired(now);
  for (const Entry& entry : expired) {
    entry.second->Run();
  }
  Reset(expired, now);
}

std::vector<TimerQueue::Entry> TimerQueue::GetExpired(Timestamp now) {
  std::vector<Entry> expired;
  auto end =
      timers_.lower_bound(Entry(now, reinterpret_cast<Timer*>(UINTPTR_MAX)));
  std::copy(timers_.begin(), end, back_inserter(expired));
  timers_.erase(timers_.begin(), end);
  return expired;  // 编译器会实施RVO返回值优化，不必太担心性能
}

void TimerQueue::Reset(const std::vector<Entry>& expired, Timestamp now) {
  for (const Entry& entry : expired) {
    Timer* timer = entry.second;
    if (timer->repeat()) {
      timer->Reset(now);
      Insert(timer);
    } else {
      delete timer;  // 可以移动到一个空闲链表，重复利用。但是短时间大量使用后，可能存在浪费
    }
  }

  if (!timers_.empty()) {
    Timestamp when_expire = timers_.begin()->first;
    if (DateTime::Valid(when_expire)) ResetTimerfd(timer_fd_, when_expire);
  }
}

bool TimerQueue::Insert(Timer* timer) {
  bool earliest_changed = false;
  auto head = timers_.begin();
  Timestamp when = timer->when_expire();
  if (head == timers_.end() || when < head->first) earliest_changed = true;

  timers_.insert(Entry(when, timer));
  return earliest_changed;
}

}  // namespace net
}  // namespace xlcomm